# -*- coding: utf-8 -*-
# @Time    : 2025/8/5 15:18
# @Author  : Zhaoxu
# @File    : together.py
# @Software: PyCharm

# -*- coding:utf-8 -*-
"""
Realtime Voice Chat  (Mic → XF ASR → GPT → XF TTS → PyAudio 播放)
-----------------------------------------------------------------
依赖:
    pip install openai websocket-client pyaudio
    # Windows 安装 PyAudio 可先:
    #   pip install pipwin
    #   pipwin install pyaudio
-----------------------------------------------------------------
用法:
    1. 麦克风说话 (最长 10 秒；静音 2.5 秒自动结束)
    2. 讯飞实时语音转写得到文本
    3. 交给 GPT 生成回答 (DeepSeek Chat 示例)
    4. GPT 答案边生成、边切块、边走讯飞 TTS 合成
    5. 每帧音频立即写入 PyAudio, 实时播放
"""
import queue
import threading
import time
import os
import base64
import json
import hashlib
import hmac
from urllib.parse import urlencode, quote
from datetime import datetime
from wsgiref.handlers import format_date_time

import websocket
import pyaudio
from websocket import create_connection
from openai import OpenAI

# ========= GPT → 队列 ========= #
def gpt_stream_to_queue(q: queue.Queue, user_prompt: str):
    """
    GPT 流式输出 → 按规则切块 → 入队
    """
    client = OpenAI(
        api_key="sk-BPVBVZQbXHrUH6b540DbE34b4c71479f9dD4106dCc29C021",#sk-9bc1ce71791b49f083e625f2f2788b2d",
        base_url="https://api.pumpkinaigc.online/v1"
    )
    resp = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "system", "content": "You are chatting with me ,your name is xiaoxi only the user use chinese speaking ,your name is 小西,reply me in English"},
            {"role": "user", "content": user_prompt}
        ],
        stream=True
    )

    buffer = ""
    for part in resp:
        delta = part.choices[0].delta
        if not getattr(delta, "content", None):
            continue
        token = delta.content
        print(token, end="", flush=True)          # 在终端同步显示
        buffer += token
        if len(buffer) >= 80 or buffer.endswith(('.', '!', '?', '。', '！', '？')):
            q.put(buffer)
            buffer = ""
    if buffer.strip():
        q.put(buffer)
    q.put(None)     # 结束信号


# ========= 讯飞 TTS websocket ========= #
class Ws_Param:
    def __init__(self, appid, apikey, apisecret, text):
        self.APPID = appid
        self.APIKey = apikey
        self.APISecret = apisecret
        self.Text = text
        self.CommonArgs = {"app_id": self.APPID}
        self.BusinessArgs = {"aue": "raw",
                             "auf": "audio/L16;rate=16000",
                             "vcn": "x4_yezi",
                             "tte": "utf8"}
        self.Data = {"status": 2,
                     "text": str(base64.b64encode(self.Text.encode('utf-8')), "utf8")}

    def create_url(self):
        url = "wss://tts-api.xfyun.cn/v2/tts"
        now = datetime.now()
        date = format_date_time(time.mktime(now.timetuple()))
        sig_origin = f"host: ws-api.xfyun.cn\n" \
                     f"date: {date}\n" \
                     f"GET /v2/tts HTTP/1.1"
        sign = hmac.new(self.APISecret.encode(), sig_origin.encode(),
                        digestmod=hashlib.sha256).digest()
        sign = base64.b64encode(sign).decode()
        auth_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", ' \
                      f'headers="host date request-line", signature="{sign}"'
        authorization = base64.b64encode(auth_origin.encode()).decode()
        return url + "?" + urlencode({"authorization": authorization,
                                      "date": date,
                                      "host": "ws-api.xfyun.cn"})


def tts_stream_and_play(appid, apikey, apisecret, text):
    """
    对 text 进行 TTS；收到音频帧立即写 PyAudio，阻塞直到播放完
    """
    p = pyaudio.PyAudio()
    stream = p.open(format=pyaudio.paInt16,
                    channels=1,
                    rate=16000,
                    output=True)

    done_flag = threading.Event()

    def on_message(ws, message):
        msg = json.loads(message)
        audio_bytes = base64.b64decode(msg["data"]["audio"])
        stream.write(audio_bytes)                  # 实时播放
        if msg["data"]["status"] == 2:
            ws.close()                             # 主动结束

    def on_error(ws, error):
        print("TTS ws error:", error)

    def on_close(ws, close_status_code, close_msg):
        stream.stop_stream()
        stream.close()
        p.terminate()
        done_flag.set()

    def on_open(ws):
        def run(*_):
            req = {"common": wsParam.CommonArgs,
                   "business": wsParam.BusinessArgs,
                   "data": wsParam.Data}
            ws.send(json.dumps(req))
        threading.Thread(target=run).start()

    wsParam = Ws_Param(appid, apikey, apisecret, text)
    websocket.enableTrace(False)
    ws = websocket.WebSocketApp(wsParam.create_url(),
                                on_open=on_open,
                                on_message=on_message,
                                on_error=on_error,
                                on_close=on_close)
    ws.run_forever(sslopt={"cert_reqs": 0})
    done_flag.wait()           # 等播放完毕


def consumer_tts_play(q: queue.Queue, appid, apikey, apisecret):
    idx = 1
    while True:
        txt = q.get()
        if txt is None:
            break
        print(f"\n\n>>> GPT 第 {idx} 段（{len(txt)} 字）开始 TTS 并播放 …")
        tts_stream_and_play(appid, apikey, apisecret, txt)
        idx += 1
    print("\n===== GPT 回答播放结束 =====")


# ========= 讯飞实时 ASR ========= #
def gen_signa(appid, apikey, ts):
    base = appid + ts
    md5 = hashlib.md5(base.encode()).hexdigest()
    signa = hmac.new(apikey.encode(), md5.encode(), digestmod='sha1').digest()
    return base64.b64encode(signa).decode()


class XFRealtimeASR:
    def __init__(self, appid, apikey, vad_eos_ms=1500):
        self.appid = appid
        self.apikey = apikey
        self.vad_eos_ms = vad_eos_ms
        self.ws = None
        self.handshake_done = threading.Event()
        self.closed = False
        self.end_tag = "{\"end\":true}"
        self.result_text = ""
        self.last_result_time = None
        self.record_start_time = None

    def connect(self):
        ts = str(int(time.time()))
        signa = gen_signa(self.appid, self.apikey, ts)
        url = f"ws://rtasr.xfyun.cn/v1/ws?appid={self.appid}&ts={ts}&signa={quote(signa)}&vad_eos={self.vad_eos_ms}"
        print("【ASR 连接URL】", url)
        self.ws = create_connection(url)
        threading.Thread(target=self._recv_thread_func, daemon=True).start()

    def start_mic(self):
        """
        开始录音 + 实时识别，返回整句文本
        """
        if not self.handshake_done.wait(10):
            print("握手失败")
            self.close()
            return ""
        CHUNK = 640
        FORMAT = pyaudio.paInt16
        CHANNELS = 1
        RATE = 16000
        pa = pyaudio.PyAudio()
        stream = pa.open(format=FORMAT, channels=CHANNELS, rate=RATE,
                         input=True, frames_per_buffer=CHUNK)
        print("【开始说话，静音 2.5 秒或 10 秒总长自动结束】")

        try:
            while not self.closed:
                data = stream.read(CHUNK, exception_on_overflow=False)
                self.ws.send(data, opcode=0x2)

                if self.record_start_time:
                    elapsed = time.time() - self.record_start_time
                    if elapsed >= 10 or (time.time() - self.last_result_time) > (self.vad_eos_ms / 1000):
                        print("【句子结束】")
                        q = queue.Queue()
                        print("大模型输入为："+self.result_text)
                        threading.Thread(target=gpt_stream_to_queue,
                                         args=(q, self.result_text), daemon=True).start()
                        XF_APPID_TTS = "9646138c"
                        XF_APIKEY_TTS = "ea89fb89877aded124af01b32cb14324"
                        XF_SECRET_TTS = "MzlhYmQyMjI5ODYzYzU2ODM2YWMxYWNk"
                        consumer_tts_play(q, XF_APPID_TTS, XF_APIKEY_TTS, XF_SECRET_TTS)
                        self.ws.send(self.end_tag.encode())
                        break
                time.sleep(0.04)
        except KeyboardInterrupt:
            print("用户中断")
        finally:
            stream.stop_stream()
            stream.close()
            pa.terminate()
            self.close()
            time.sleep(0.2)
            return self.result_text

    def parse_data_text(self, data_str):
        try:
            data_dict = json.loads(data_str)
            words = []
            for rt in data_dict.get('cn', {}).get('st', {}).get('rt', []):
                for ws in rt['ws']:
                    for cw in ws['cw']:
                        words.append(cw['w'])
            return ''.join(words)
        except Exception as e:
            print("解析 ASR 文本出错:", e)
            return ""

    def _recv_thread_func(self):
        try:
            while self.ws.connected:
                result = self.ws.recv()
                if not result:
                    break
                j = json.loads(result)
                if j["action"] == "started":
                    self.handshake_done.set()
                elif j["action"] == "result":
                    txt = self.parse_data_text(j.get("data", ""))
                    if txt:
                        print(f"【实时识别】{txt}")
                        self.result_text = txt
                        self.last_result_time = time.time()
                        if self.record_start_time is None:
                            self.record_start_time = time.time()
                elif j["action"] == "error":
                    print("ASR 错误:", j)
                    self.closed = True
                    self.ws.close()
                    break
        except Exception as e:
            if not self.closed:
                print("ASR 接收线程异常:", e)

    def close(self):
        self.closed = True
        try:
            if self.ws:
                self.ws.close()
                print("ASR WebSocket 关闭")
        except:
            pass


# ========= main ========= #
def main():
    # ======= 讯飞账号 =======    appid = "9646138c"
    #     apikey = "82adf64eb83cf7b9ec6f4ae215943e0d"
    XF_APPID_ASR   = "9646138c"
    XF_APIKEY_ASR  = "82adf64eb83cf7b9ec6f4ae215943e0d"

    # APPID = "9646138c"
    # APIKEY = "ea89fb89877aded124af01b32cb14324"
    # APISECRET = "MzlhYmQyMjI5ODYzYzU2ODM2YWMxYWNk"
    # ---- Step1: 语音实时转写 ---- #
    asr = XFRealtimeASR(XF_APPID_ASR, XF_APIKEY_ASR, vad_eos_ms=1500)
    asr.connect()
    user_text = asr.start_mic()
    print("\n【最终识别文本】", user_text)

    if not user_text.strip():
        print("未识别到文本，结束")
        return

if __name__ == "__main__":
    main()
